package com.my.thread.pool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by ZhongGang
 * at 2018/7/13 10:10
 * <p>
 * 线程池实现
 * <p>
 * 问题一： 如何实现线程池的关闭
 * 问题二： 如何实现线程池工作线程大小的动态扩展与收缩
 */
public class ThreadPool {
    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPool.class);
    private static final int PROCESSORS = Runtime.getRuntime().availableProcessors();
    private static final AtomicInteger POOL_INDEX = new AtomicInteger(0);//全局pool标识

    private int min = PROCESSORS;//最少线程执行数量
    private int max = PROCESSORS;//最多线程执行数量，过多的线程数量会导致系统频繁的在线程间进行切换，浪费资源，默认设置成系统的CPU核心数
    private BlockingQueue<Task> tasks;//装载任务线程对象的队列
    private long timeout = 1L;//扩展线程空闲时最大时间
    private TimeUnit unit = TimeUnit.MINUTES;//扩展线程空闲时最大时间单位
    private ThreadFactory threadFactory = new DefaultThreadFactory();
    private boolean shutdown = false;//线程池是否关闭标志符
    private int currentPoolIndex = POOL_INDEX.getAndIncrement();//当前线程池对象标识
    private List<Thread> threads = new ArrayList<>();//当前正在运行的线程
    private ReentrantLock lock = new ReentrantLock();

    public ThreadPool(BlockingQueue<Task> tasks) {
        this(PROCESSORS, PROCESSORS, tasks);
    }

    public ThreadPool(int min, int max, BlockingQueue<Task> tasks) {
        if (min <= 0) {
            throw new IllegalArgumentException("min must be greater then 0.");
        }
        if (max <= 0) {
            throw new IllegalArgumentException("max must be greater then 0.");
        }
        this.min = min;
        this.max = max;
        this.tasks = tasks;
    }

    public <V> AsyncResult<V> submit(Runnable runnable) throws Exception {
        try {
            this.lock.lock();
            AsyncResult<V> asyncResult = new AsyncResult<>();
            RunnableTask<V> task = new RunnableTask<>(runnable, asyncResult);
            addTask(task);
            return asyncResult;
        } finally {
            this.lock.unlock();
        }
    }

    public <V> AsyncResult<V> submit(Runnable runnable, V result) throws Exception {
        try {
            this.lock.lock();
            AsyncResult<V> asyncResult = new AsyncResult<>(result);
            RunnableTask<V> task = new RunnableTask<>(runnable, asyncResult);
            addTask(task);
            return asyncResult;
        } finally {
            this.lock.unlock();
        }
    }

    public <V> AsyncResult<V> submit(Callable<V> callable) throws Exception {
        try {
            this.lock.lock();
            AsyncResult<V> asyncResult = new AsyncResult<>();
            CallableTask<V> task = new CallableTask<>(callable, asyncResult);
            addTask(task);
            return asyncResult;
        } finally {
            this.lock.unlock();
        }
    }

    private <V> void addTask(Task<V> task) {
        if (this.shutdown) {
            throw new IllegalStateException("has been shutdown, can not submit any tasks...");
        }

        if (this.threads.size() < this.min) {
            createWorkTask();
        }
        boolean notFull = this.tasks.offer(task);

        if (!notFull && this.min < this.max && this.threads.size() < this.max) {
            createExtendWorkTask();
        }
    }

    private void createWorkTask() {
        Thread thread = this.threadFactory.create(new Worker(this.tasks));
        thread.setName(String.format("pool-%s-%s", this.currentPoolIndex, thread.getName()));
        this.threads.add(thread);
        thread.start();
    }

    private void createExtendWorkTask() {
        Thread thread = this.threadFactory.create(new ExtendWorker(this.tasks, this.threads, this.timeout, this.unit));
        thread.setName(String.format("pool-%s-extend-%s", this.currentPoolIndex, thread.getName()));
        this.threads.add(thread);
        thread.start();
    }

    public void shutdown() {
        try {
            this.lock.lock();

            this.shutdown = true;

            for (Thread thread : threads) {
                thread.interrupt();
            }
        } finally {
            this.lock.unlock();
        }
    }
}
